package com.edu360.select

import com.edu360.utils.{AreaDistributionUtil, NBF, ToMysqlUtils}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object SelectMediaSQL {
  def main(args: Array[String]): Unit = {
    // 0 校验参数个数
    if (args.length != 3) {
      println(
        """
          |cn.dmp.tools.Bzip2Parquet
          |参数：
          | logInputPath
          | ruleInputPath
          | resultOutputPath
        """.stripMargin)
      sys.exit()
    }
    // 1 接受程序参数
    val Array(logInputPath,ruleInputPath,resultOutputPath) = args
    // 2 创建sparkconf->sparkContext
    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")
    val sc = new SparkContext(sparkConf)
    // RDD 序列化到磁盘 worker与worker之间的数据传输
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sQLContext = new SQLContext(sc)

    //取到HDFS中的app规则
    val rulesLines:RDD[String] = sc.textFile(ruleInputPath)

    //整理app规则数据
    val appRulesRDD: RDD[(String, String)] = rulesLines.map(line => {
      val fields = line.split("\t")
      val appId = fields(4)
      val appName = fields(1)
      (appId, appName)
    })

    //将分散在多个Executor中的部分IP规则收集到Driver端
    val rulesInDriver = appRulesRDD.collect().toMap

    //将Driver端的数据广播到Executor
    //广播变量的引用（还在Driver端）
    val broadcastRef= sc.broadcast(rulesInDriver)


    //读取parquet文件
    val parquet: DataFrame = sQLContext.read.parquet(logInputPath)
    ////把DataFrame先注册临时表
    parquet.registerTempTable("logs")

    sQLContext.udf.register("searchAppname",(appid:String,appname:String)=>{
      //使用广播变量的引用，就可以获得
      val rules= broadcastRef.value
      var app=appname
      if ("".equals(app.trim())||"未知".equals(app.trim())||"其他".equals(app.trim())){
        app=rules.getOrElse(appid,"无名氏")
      }
      app
    })
    sQLContext.udf.register("selPriReq",AreaDistributionUtil.selectPrimaryRequest _)
    sQLContext.udf.register("selectValidRequest",AreaDistributionUtil.selectValidRequest _)
    sQLContext.udf.register("selectAdvRequest",AreaDistributionUtil.selectAdvRequest _)
    sQLContext.udf.register("selectJoinBidding",AreaDistributionUtil.selectJoinBidding _)
    sQLContext.udf.register("selectSuccessBidding",AreaDistributionUtil.selectSuccessBidding _)
    sQLContext.udf.register("selectAdverShow",AreaDistributionUtil.selectAdverShow _)
    sQLContext.udf.register("selectAdverClick",AreaDistributionUtil.selectAdverClick _)
    sQLContext.udf.register("selectDspConsume",AreaDistributionUtil.selectDspConsume _)
    sQLContext.udf.register("selectDspCost",AreaDistributionUtil.selectDspCost _)
    //计算原始请求
    val sql = sQLContext.sql("select searchAppname(appid,appname) media,sum(selPriReq(requestmode,processnode)) 原始请求数,sum(selectValidRequest(requestmode,processnode)) 有效请求数,sum(selectAdvRequest(requestmode,processnode)) 广告请求数,sum(selectJoinBidding(iseffective,isbilling,isbid,adorderid)) 参与竞价数,sum(selectSuccessBidding(iseffective,isbilling,iswin)) 竞价成功数,sum(selectAdverShow(requestmode,iseffective)) 展示数,sum(selectAdverClick(requestmode,iseffective)) 点击数,sum(selectDspConsume(iseffective,isbilling,iswin,winprice)) DSP广告消费,sum(selectDspCost(iseffective,isbilling,iswin,adpayment)) DSP广告成本  from logs group by searchAppname(appid,appname)")
    ToMysqlUtils.dfToSql(sql,resultOutputPath)
    sc.stop()
  }
}
